package com.learn.kafka.idempotence;

import com.learn.kafka.config.ProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * 幂等性生产者，继承 ProducerProperties 复用基础配置
 * @author: learn.kafka
 */
public class IdempotenceProducer extends ProducerProperties {


    public static void main(String[] args) {
        // 创建幂等性生产者实例
        IdempotenceProducer producer = new IdempotenceProducer();
        producer.startMessages();
    }

    // 构造方法：初始化幂等性配置
    public IdempotenceProducer() {
        super(); // 调用父类构造方法，加载基础配置
        // 在父类基础配置上添加幂等性相关配置
        initIdempotenceConfig();
    }

    // 初始化幂等性配置
    private void initIdempotenceConfig() {
        // 开启幂等性（必须设置为true）
        super.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

        // 幂等性建议配合以下配置（若父类默认值不满足需求可覆盖）
        // 1. acks建议设置为all（最高可靠性，父类若为1可覆盖）
        super.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
        // 2. 重试次数建议适当增大（父类默认3，可根据需求调整）
        super.setProperty(ProducerConfig.RETRIES_CONFIG, "5");
    }

    // 发送消息的方法
    private void startMessages() {
        // 调用父类方法创建生产者，消除重复代码
        Producer<Integer, String> producer = super.createProducer();

        // 发送10条测试消息
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("mytopic", i, Integer.toString(i)));
        }

        // 关闭生产者
        producer.close();
    }
}
